# Join employment status to linked trips
employment_linked_trips = linked_trips.join(
persons_df.select(["person_id", "employment"]),
on="person_id",
how="left",
)
# Calculate total weighted and unweighted days by employment status
employment_day_totals = (
days_df
.join(
persons_df.select(["person_id", "employment", "hh_id"]),
on=["person_id", "hh_id"],
how="left",
)
.group_by("employment").agg([
pl.len().alias("unweighted_days"),
pl.col("day_weight").sum().alias("weighted_days"),
])
.sort("unweighted_days", descending=True)
)
# Calculate total work and work-related trips by employment status
employment_work_trips = (
employment_linked_trips
.filter(
pl.col("d_purpose_category").is_in([2,3]) &
(pl.col("linked_trip_weight") > 0), # Drop incomplete trips
)
.group_by(["employment", "d_purpose_category"]).agg([
pl.len().alias("unweighted_work_trips"),
pl.col("linked_trip_weight").sum().alias("weighted_work_trips"),
])
.with_columns(
pl.col("employment").cast(pl.Utf8).replace(EMPLOY_MAP).alias("employment_status"),
pl.col("d_purpose_category").cast(pl.Utf8).replace(PURPOSE_MAP).alias("purpose"),
)
.sort("unweighted_work_trips", descending=True)
)
# Merge and calculate trip rates
employment_summary = (
employment_day_totals
# Join work trips
.join(
employment_work_trips,
on="employment",
how="left",
)
)
# Calculate trip rates
employment_summary = (
employment_summary
.with_columns([
(pl.col("unweighted_work_trips") / pl.col("unweighted_days"))
.alias("unweighted_trip_rate"),
(pl.col("weighted_work_trips") / pl.col("weighted_days"))
.alias("weighted_trip_rate"),
])
)
# Just display Employed and Self-employed
emp_plot_dat = employment_summary.filter(
pl.col("employment").is_in([1,2,3,7])
).drop(["d_purpose_category", "employment"])
fig = go.Figure()
# Get unique purposes and employment statuses
purposes = emp_plot_dat["purpose"].unique().to_list()
employment_statuses = emp_plot_dat["employment_status"].unique().to_list()
colors = px.colors.qualitative.Plotly
# For each purpose, create grouped bars
for i, purpose in enumerate(purposes):
purpose_data = emp_plot_dat.filter(pl.col("purpose") == purpose)
# Add weighted bars (solid)
fig.add_trace(go.Bar(
name=f"{purpose} (Weighted)",
x=purpose_data["employment_status"],
y=purpose_data["weighted_trip_rate"],
marker_color=colors[i % len(colors)],
text=purpose_data["weighted_trip_rate"].round(2),
texttemplate="%{text:.2f}",
textposition="inside",
legendgroup=purpose,
offsetgroup=i,
))
# Add unweighted bars (hatched overlay) on top of the same bars
fig.add_trace(go.Bar(
name=f"{purpose} (Unweighted)",
x=purpose_data["employment_status"],
y=purpose_data["unweighted_trip_rate"],
marker_color=colors[i % len(colors)],
marker_pattern_shape="/",
marker_pattern_solidity=0.3,
opacity=0.7,
legendgroup=purpose,
offsetgroup=i,
text=purpose_data["unweighted_trip_rate"].round(2),
texttemplate="%{text:.2f}",
textposition="inside",
))
fig.update_layout(
barmode="group",
title="Weekday Work Trip Rates by Employment",
xaxis_title="Employment Status",
yaxis_title="Trip Rate (Trips per Day)",
yaxis_range=[0, 0.6],
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1)
)
fig.show()